import concurrent.futures
import datetime

import pandas as pd
from click.core import batch
from retrying import retry
from tqdm import tqdm
from loguru import logger

from apps.pdd.pdd.last_deal.export import LastDataExport
from apps.pdd.pdd.last_deal.trend_mall import TrendMall
from components.config import WFQ_SOURCE_MYSQL_CONFIG
from utils.db.mysqldb import MysqlDB


def run_core(month, mall_ids):
    df = TrendMall(month).run_one(mall_ids)
    # df = pd.DataFrame()
    LastDataExport().run_one(mall_ids)
    return df


class ResetAnalysis:
    def __init__(self):
        self._to_db = None

    @staticmethod
    def get_date_list_items(analysis_version_flag):
        item_list = [
            {
                "analysis_version": "202501",
                "analysis_version_date": [f"2024-12-{i}" for i in range(25, 32)]
                + [f"2025-01-{str(i).zfill(2)}" for i in range(1, 11)],
            }
        ]
        for j in range(2, 13):
            analysis_version = f"2025{str(j).zfill(2)}"
            analysis_version_date = [f"2025-{str(j - 1).zfill(2)}-{i}" for i in range(25, 32)] + [
                f"2025-{str(j).zfill(2)}-{str(i).zfill(2)}" for i in range(1, 16)
            ]
            item_list.append({"analysis_version": analysis_version, "analysis_version_date": analysis_version_date})
        print(item_list)
        for item in item_list:
            if item["analysis_version"] == analysis_version_flag:
                return item["analysis_version_date"]
        return item_list

    @property
    def to_db(self):
        if self._to_db is None:
            self._to_db = MysqlDB(
                ip=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_IP"],
                port=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_PORT"],
                db=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_DB"],
                user_name=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_NAME"],
                user_pass=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_PASS"],
            )
        return self._to_db

    def gen_query_params(self, month):
        date_list = self.get_date_list_items(month)
        date_list_str = ",".join([f"'{i}'" for i in date_list])
        sql = """
        select mall_id, max(version) as version
        from source_net_pdd_shop_info
        where shop_name != '' and shop_name is not null
        and mall_id in (select mall_id from zm_pdd_mall_id)
         group by mall_id
        """
        print(sql)
        query_params = self.to_db.find(sql, to_json=True)
        sql_list = [f" (mall_id = '{i['mall_id']}' and version = '{i['version']}') " for i in query_params]
        return sql_list

    @retry(stop_max_attempt_number=5, wait_fixed=5000)
    def find_by_sql(self, sql):
        return self.to_db.find(sql, to_json=True)

    def get_mall_data(self, params_list):
        batch_size = 100
        datas = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(
                    self.find_by_sql,
                    f"""select * from source_net_pdd_shop_info where { " or ".join(params_list[i:i+batch_size]) }""",
                )
                for i in range(0, len(params_list), batch_size)
            ]
            for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures)):
                datas.extend(future.result())
        return pd.DataFrame(datas)

    def get_goods_data(self, params_list):
        batch_size = 100
        datas = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(
                    self.find_by_sql,
                    f"""select * from source_result_pdd_goods_detail where { " or ".join(params_list[i:i+batch_size]) }""",
                )
                for i in range(0, len(params_list), batch_size)
            ]
            for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures)):
                datas.extend(future.result())
        return pd.DataFrame(datas)

    def gen_data(self, month):
        params = self.gen_query_params(month)
        mall_data = self.get_mall_data(params)
        # goods_data = self.get_goods_data(params)
        # mall_data.to_csv(f"mall_data_{month}.csv")
        # goods_data.to_csv(f"goods_data_{month}.csv")
        # return mall_data, goods_data
        return mall_data, pd.DataFrame()

    def run(self):
        # with pd.ExcelWriter(f"trend_mall_{datetime.datetime.now().strftime('%Y%m%d%H%M')}.xlsx") as writer:
        for month in [
            "202506",
            # "202503",
            # "202504"
        ]:
            mall_data, goods_data = self.gen_data(month)
            mall_ids = mall_data["mall_id"].to_list()
            logger.info(f"处理店铺数量{len(mall_ids)}")
            batch_size = 10
            with concurrent.futures.ProcessPoolExecutor(max_workers=10) as executor:
                futures = [
                    executor.submit(run_core, **{"month": month, "mall_ids": mall_ids[i:i + batch_size]})
                    for i in range(0, len(mall_ids), batch_size)
                ]
                for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures)):
                    if future.exception():
                        raise future.exception()
                    result = future.result()
            #         if result:
            #             df = pd.concat([df, pd.DataFrame(result)])
            # # df = TrendMall(month).run(mall_ids)
            # df.to_excel(writer, sheet_name=month)
            # try:
            #     print(month, df.shape, df['month_sales'].sum(), df['month_sales_money'].sum(), df['month_sales_money'].sum() / df.shape[0] / 30)
            # except Exception as e:
            #     print(e)


class InCreateResetAnalysis(ResetAnalysis):
    @staticmethod
    def get_date_list_items(analysis_version_flag):
        return [datetime.datetime.now().strftime("%Y-%m-%d")]

    def run(self):
        super().run()
        sql = f"""
        select count(*) as count
        from zm_pdd_mall_id
        where mall_id not in (
            select distinct zm_result_pdd_shop_info.mall_id
            from zm_result_pdd_shop_info where version >= '{(datetime.datetime.now() - datetime.timedelta(days=30)).strftime("%Y-%m")}-25'
        )
        """
        print(sql)
        data = self.to_db.find(sql=sql, to_json=True)
        print(data)


if __name__ == "__main__":
    # mall_ids = ['328636']
    # mall_ids = ['238355015']
    # batch_size = 10
    # with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
    #     futures = [
    #         executor.submit(
    #             run_core,
    #             month="202506",
    #             mall_ids=mall_ids[i:i + batch_size],
    #         )
    #         for i in range(0, len(mall_ids), batch_size)
    #     ]
    #     for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures)):
    #         if future.exception():
    #             raise future.exception()

    ResetAnalysis().run()
    # print(ResetAnalysis().get_date_list_items('202506'))

    #
    # InCreateResetAnalysis().run()
    # from apscheduler.schedulers.blocking import BlockingScheduler
    # sched = BlockingScheduler()
    # # 添加任务（支持cron表达式）
    # sched.add_job(InCreateResetAnalysis().run, "interval", minutes=5, next_run_time=datetime.datetime.now())  # 间隔执行
    #
    # try:
    #     sched.start()
    # except KeyboardInterrupt:
    #     sched.shutdown()
